name: pre-commit run_id: commands[2] env HOME: /home/jenkins env LANG: C.UTF-8 env OLM_TIMER1: 3000 env OLM_TIMER2: 2000 env PATH: /w/workspace/transportpce-tox-verify-transportpce-master/.tox/pre-commit/bin:/opt/pyenv/bin:/tmp/venv-Pb6O/bin:/opt/pyenv/shims:/home/jenkins/.local/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/usr/games:/usr/local/games:/snap/bin:/opt/puppetlabs/bin env PIP_DISABLE_PIP_VERSION_CHECK: 1 env PYTHONHASHSEED: 4169902129 env PYTHONIOENCODING: utf-8 env SSH_AUTH_SOCK: ******************************** env TOX_ENV_DIR: /w/workspace/transportpce-tox-verify-transportpce-master/.tox/pre-commit env TOX_ENV_NAME: pre-commit env TOX_WORK_DIR: /w/workspace/transportpce-tox-verify-transportpce-master/.tox env VIRTUAL_ENV: /w/workspace/transportpce-tox-verify-transportpce-master/.tox/pre-commit env __TOX_ENVIRONMENT_VARIABLE_ORIGINAL_CI: true metadata pid: 2494 cwd: /w/workspace/transportpce-tox-verify-transportpce-master/tests allow: /w/workspace/transportpce-tox-verify-transportpce-master/.tox/pre-commit/bin/*:sh:cpan:sudo:which:fixCIcentOS8reposMirrors.sh cmd: pre-commit run --all-files --show-diff-on-failure exit_code: 1 [WARNING] hook id `remove-tabs` uses deprecated stage names (commit) which will be removed in a future version. run: `pre-commit migrate-config` to automatically fix this. [WARNING] hook id `perltidy` uses deprecated stage names (commit) which will be removed in a future version. run: `pre-commit migrate-config` to automatically fix this. [INFO] Initializing environment for https://github.com/pre-commit/pre-commit-hooks. [WARNING] repo `https://github.com/pre-commit/pre-commit-hooks` uses deprecated stage names (commit, push) which will be removed in a future version. Hint: often `pre-commit autoupdate --repo https://github.com/pre-commit/pre-commit-hooks` will fix this. if it does not -- consider reporting an issue to that repo. [INFO] Initializing environment for https://github.com/jorisroovers/gitlint. [INFO] Initializing environment for https://github.com/jorisroovers/gitlint:./gitlint-core[trusted-deps]. [INFO] Initializing environment for https://github.com/Lucas-C/pre-commit-hooks. [INFO] Initializing environment for https://github.com/pre-commit/mirrors-autopep8. [INFO] Initializing environment for https://github.com/perltidy/perltidy. [INFO] Installing environment for https://github.com/pre-commit/pre-commit-hooks. [INFO] Once installed this environment will be reused. [INFO] This may take a few minutes... [INFO] Installing environment for https://github.com/Lucas-C/pre-commit-hooks. [INFO] Once installed this environment will be reused. [INFO] This may take a few minutes... [INFO] Installing environment for https://github.com/pre-commit/mirrors-autopep8. [INFO] Once installed this environment will be reused. [INFO] This may take a few minutes... [INFO] Installing environment for https://github.com/perltidy/perltidy. [INFO] Once installed this environment will be reused. [INFO] This may take a few minutes... trim trailing whitespace.................................................Failed - hook id: trailing-whitespace - exit code: 1 - files were modified by this hook Fixing tests/transportpce_tests/pce/test05_pce_tapi.py Fixing tapi/src/main/java/org/opendaylight/transportpce/tapi/connectivity/ConnectivityUtils.java Fixing tests/transportpce_tests/common/test_utils_generate_tapi_topo.py Tabs remover.............................................................Passed autopep8.................................................................Failed - hook id: autopep8 - files were modified by this hook perltidy.................................................................Passed pre-commit hook(s) made changes. If you are seeing this message in CI, reproduce locally with: `pre-commit run --all-files`. To run `pre-commit` as part of git workflow, use `pre-commit install`. All changes made by hooks: diff --git a/tapi/src/main/java/org/opendaylight/transportpce/tapi/connectivity/ConnectivityUtils.java b/tapi/src/main/java/org/opendaylight/transportpce/tapi/connectivity/ConnectivityUtils.java index a93733027..62b6e8f66 100644 --- a/tapi/src/main/java/org/opendaylight/transportpce/tapi/connectivity/ConnectivityUtils.java +++ b/tapi/src/main/java/org/opendaylight/transportpce/tapi/connectivity/ConnectivityUtils.java @@ -847,7 +847,7 @@ public final class ConnectivityUtils { input.getConnectivityConstraint().getRequestedCapacity().getTotalSize().getValue().intValue())); // map endpoints into service end points. Map the type of service from TAPI to OR LOG.info("Calling tapiEndPointToServiceAPoint w endpoint {}, ServiceFormat = {}, nodeAId = {}, capacity = {}," - + " input LayerProtocolName = {}", endPointMap.values().stream().findFirst().orElseThrow(), + + " input LayerProtocolName = {}", endPointMap.values().stream().findFirst().orElseThrow(), serviceFormat, nodeAAid, capacity, input.getLayerProtocolName()); ServiceAEnd serviceAEnd = tapiEndPointToServiceAPoint( endPointMap.values().stream().findFirst().orElseThrow(), diff --git a/tests/transportpce_tests/common/test_utils_generate_tapi_topo.py b/tests/transportpce_tests/common/test_utils_generate_tapi_topo.py index 375b1b4ec..40d816398 100644 --- a/tests/transportpce_tests/common/test_utils_generate_tapi_topo.py +++ b/tests/transportpce_tests/common/test_utils_generate_tapi_topo.py @@ -24,6 +24,7 @@ sys.path.append('transportpce_tests/common/') import test_utils # nopep8 # pylint: disable=too-few-public-methods + class UuidServices: def __init__(self): # pylint: disable=invalid-name @@ -31,6 +32,7 @@ class UuidServices: self.odu = None self.dsr = None + class UuidServices2: def __init__(self): # pylint: disable=invalid-name @@ -38,6 +40,7 @@ class UuidServices2: self.odu = None self.dsr = None + class UuidServices3: def __init__(self): # pylint: disable=invalid-name @@ -47,6 +50,7 @@ class UuidServices3: # class TransportPCEtesting(unittest.TestCase): + processes = None WAITING = 20 # nominal value is 300 NODE_VERSION = '2.2.1' @@ -56,6 +60,7 @@ uuid_services3 = UuidServices3() del_serv_input_data = {"uuid": "TBD"} tapi_topo = {"topology-id": "TBD"} + def initSimulators(): print("Entering TestUtils setupClass starting simulators") print("Node version is {}".format(NODE_VERSION)) @@ -66,8 +71,9 @@ def initSimulators(): ('xpdrc', NODE_VERSION), ('spdrc', NODE_VERSION)]) + def connect_xpdrs(): - + print("Connecting SPDRA") response = test_utils.mount_device("SPDR-SA1", ('spdra', NODE_VERSION)) time.sleep(2) @@ -81,6 +87,7 @@ def connect_xpdrs(): response = test_utils.mount_device("XPDR-C1", ('xpdrc', NODE_VERSION)) time.sleep(2) + def connect_rdms(): print("Connecting ROADMA") response = test_utils.mount_device("ROADM-A1", ('roadma', NODE_VERSION)) @@ -93,6 +100,7 @@ def connect_rdms(): print("Connecting ROADMC failed") time.sleep(2) + def interconnect_roadm_and_xpdr(): response = test_utils.transportpce_api_rpc_request( 'transportpce-networkutils', 'init-xpdr-rdm-links', @@ -295,6 +303,7 @@ def interconnect_roadm_and_xpdr(): print("Connection from ROADMC1 to SPDRSC1-X2 succeeded") time.sleep(2) + def add_omsAttributes_to_ROADM2ROADM_links(): # Config ROADMA-ROADMC oms-attributes data = {"span": { @@ -331,6 +340,7 @@ def add_omsAttributes_to_ROADM2ROADM_links(): else: print("OMS attributes added on link ROADMC-ROADMA") + def create_services_on_infra(self): create_connectivity_service_PhotonicMedia(self) get_service_PhotonicMedia(self) @@ -345,6 +355,7 @@ def create_services_on_infra(self): create_connectivity_service_DSR(self) get_service_DSR(self) + def create_connectivity_service_PhotonicMedia(self): self.cr_serv_input_data["end-point"][0]["service-interface-point"]["service-interface-point-uuid"] = self.sAOTS self.cr_serv_input_data["end-point"][1]["service-interface-point"]["service-interface-point-uuid"] = self.sZOTS @@ -371,14 +382,15 @@ def create_connectivity_service_PhotonicMedia(self): 'value': 'SPDR-SA1-XPDR1'} tpcetest.assertDictEqual(dict(input_dict_1, **response['output']['service']), - response['output']['service']) + response['output']['service']) tpcetest.assertDictEqual(dict(input_dict_2, **response['output']['service']['end-point'][0]['name'][0]), - response['output']['service']['end-point'][0]['name'][0]) + response['output']['service']['end-point'][0]['name'][0]) tpcetest.assertDictEqual(dict(input_dict_3, **response['output']['service']['end-point'][1]['name'][0]), - response['output']['service']['end-point'][1]['name'][0]) + response['output']['service']['end-point'][1]['name'][0]) # If the gate fails is because of the waiting time not being enough # time.sleep(self.WAITING) + def get_service_PhotonicMedia(self): response = test_utils.get_ordm_serv_list_attr_request("services", str(tpcetest.uuid_services.pm)) print("getServiceResponse Service1 PhotonicMedia: {}".format(response)) @@ -389,6 +401,7 @@ def get_service_PhotonicMedia(self): tpcetest.assertEqual(response['services'][0]['lifecycle-state'], 'planned') time.sleep(1) + def create_connectivity_service2_PhotonicMedia(self): self.cr_serv_input_data["end-point"][0]["service-interface-point"]["service-interface-point-uuid"] = self.s2AOTS self.cr_serv_input_data["end-point"][1]["service-interface-point"]["service-interface-point-uuid"] = self.s2ZOTS @@ -416,14 +429,15 @@ def create_connectivity_service2_PhotonicMedia(self): 'value': 'SPDR-SA1-XPDR2'} tpcetest.assertDictEqual(dict(input_dict_1, **response['output']['service']), - response['output']['service']) + response['output']['service']) tpcetest.assertDictEqual(dict(input_dict_2, **response['output']['service']['end-point'][0]['name'][0]), - response['output']['service']['end-point'][0]['name'][0]) + response['output']['service']['end-point'][0]['name'][0]) tpcetest.assertDictEqual(dict(input_dict_3, **response['output']['service']['end-point'][1]['name'][0]), - response['output']['service']['end-point'][1]['name'][0]) + response['output']['service']['end-point'][1]['name'][0]) # If the gate fails is because of the waiting time not being enough # time.sleep(self.WAITING) + def get_service2_PhotonicMedia(self): response = test_utils.get_ordm_serv_list_attr_request("services", str(tpcetest.uuid_services2.pm)) print("getServiceResponse Service2 PhotonicMedia: {}".format(response)) @@ -434,6 +448,7 @@ def get_service2_PhotonicMedia(self): tpcetest.assertEqual(response['services'][0]['lifecycle-state'], 'planned') time.sleep(1) + def create_connectivity_service3_PhotonicMedia(self): self.cr_serv_input_data["end-point"][0]["service-interface-point"]["service-interface-point-uuid"] = self.s3AOTS self.cr_serv_input_data["end-point"][1]["service-interface-point"]["service-interface-point-uuid"] = self.s3ZOTS @@ -461,14 +476,15 @@ def create_connectivity_service3_PhotonicMedia(self): 'value': 'SPDR-SA1-XPDR3'} tpcetest.assertDictEqual(dict(input_dict_1, **response['output']['service']), - response['output']['service']) + response['output']['service']) tpcetest.assertDictEqual(dict(input_dict_2, **response['output']['service']['end-point'][0]['name'][0]), - response['output']['service']['end-point'][0]['name'][0]) + response['output']['service']['end-point'][0]['name'][0]) tpcetest.assertDictEqual(dict(input_dict_3, **response['output']['service']['end-point'][1]['name'][0]), - response['output']['service']['end-point'][1]['name'][0]) + response['output']['service']['end-point'][1]['name'][0]) # If the gate fails is because of the waiting time not being enough # time.sleep(self.WAITING) + def get_service3_PhotonicMedia(self): response = test_utils.get_ordm_serv_list_attr_request("services", str(tpcetest.uuid_services3.pm)) print("getServiceResponse Service3 PhotonicMedia: {}".format(response)) @@ -480,6 +496,8 @@ def get_service3_PhotonicMedia(self): time.sleep(1) # test create connectivity service from spdrA to spdrC for odu + + def create_connectivity_service_ODU(self): # pylint: disable=line-too-long self.cr_serv_input_data["layer-protocol-name"] = "ODU" @@ -515,14 +533,15 @@ def create_connectivity_service_ODU(self): 'value': 'SPDR-SA1-XPDR1'} tpcetest.assertDictEqual(dict(input_dict_1, **response['output']['service']), - response['output']['service']) + response['output']['service']) tpcetest.assertDictEqual(dict(input_dict_2, **response['output']['service']['end-point'][0]['name'][0]), - response['output']['service']['end-point'][0]['name'][0]) + response['output']['service']['end-point'][0]['name'][0]) tpcetest.assertDictEqual(dict(input_dict_3, **response['output']['service']['end-point'][1]['name'][0]), - response['output']['service']['end-point'][1]['name'][0]) + response['output']['service']['end-point'][1]['name'][0]) # If the gate fails is because of the waiting time not being enough # time.sleep(self.WAITING) + def get_service_ODU(self): response = test_utils.get_ordm_serv_list_attr_request("services", str(tpcetest.uuid_services.odu)) tpcetest.assertEqual(response['status_code'], requests.codes.ok) @@ -533,6 +552,8 @@ def get_service_ODU(self): time.sleep(1) # test create connectivity service from spdrA to spdrC for odu + + def create_connectivity_service3_ODU(self): # pylint: disable=line-too-long self.cr_serv_input_data["layer-protocol-name"] = "ODU" @@ -568,14 +589,15 @@ def create_connectivity_service3_ODU(self): 'value': 'SPDR-SA1-XPDR3'} tpcetest.assertDictEqual(dict(input_dict_1, **response['output']['service']), - response['output']['service']) + response['output']['service']) tpcetest.assertDictEqual(dict(input_dict_2, **response['output']['service']['end-point'][0]['name'][0]), - response['output']['service']['end-point'][0]['name'][0]) + response['output']['service']['end-point'][0]['name'][0]) tpcetest.assertDictEqual(dict(input_dict_3, **response['output']['service']['end-point'][1]['name'][0]), - response['output']['service']['end-point'][1]['name'][0]) + response['output']['service']['end-point'][1]['name'][0]) # If the gate fails is because of the waiting time not being enough # time.sleep(self.WAITING) + def get_service3_ODU(self): response = test_utils.get_ordm_serv_list_attr_request("services", str(tpcetest.uuid_services3.odu)) tpcetest.assertEqual(response['status_code'], requests.codes.ok) @@ -585,6 +607,8 @@ def get_service3_ODU(self): tpcetest.assertEqual(response['services'][0]['lifecycle-state'], 'planned') # test create connectivity service from spdrA to spdrC for dsr + + def create_connectivity_service_DSR(self): # pylint: disable=line-too-long self.cr_serv_input_data["layer-protocol-name"] = "DSR" @@ -621,17 +645,18 @@ def create_connectivity_service_DSR(self): 'value': 'SPDR-SA1-XPDR1'} tpcetest.assertDictEqual(dict(input_dict_1, - **response['output']['service']), - response['output']['service']) + **response['output']['service']), + response['output']['service']) tpcetest.assertDictEqual(dict(input_dict_2, - **response['output']['service']['end-point'][0]['name'][0]), - response['output']['service']['end-point'][0]['name'][0]) + **response['output']['service']['end-point'][0]['name'][0]), + response['output']['service']['end-point'][0]['name'][0]) tpcetest.assertDictEqual(dict(input_dict_3, - **response['output']['service']['end-point'][1]['name'][0]), - response['output']['service']['end-point'][1]['name'][0]) + **response['output']['service']['end-point'][1]['name'][0]), + response['output']['service']['end-point'][1]['name'][0]) # The sleep here is okey as the DSR service creation is very fast # time.sleep(self.WAITING) + def get_service_DSR(self): response = test_utils.get_ordm_serv_list_attr_request("services", str(tpcetest.uuid_services.dsr)) tpcetest.assertEqual(response['status_code'], requests.codes.ok) @@ -643,6 +668,7 @@ def get_service_DSR(self): print("Time to retrieve topology : 100 seconds... Hurry up") time.sleep(100) + def delete_created_tapi_services(self): delete_connectivity_service_DSR(self) delete_connectivity_service_ODU(self) @@ -652,6 +678,7 @@ def delete_created_tapi_services(self): delete_connectivity_service_PhotonicMedia(self) get_no_tapi_services() + def delete_connectivity_service_DSR(self): self.del_serv_input_data["uuid"] = str(tpcetest.uuid_services.dsr) response = test_utils.transportpce_api_rpc_request( @@ -659,6 +686,7 @@ def delete_connectivity_service_DSR(self): tpcetest.assertIn(response["status_code"], (requests.codes.ok, requests.codes.no_content)) time.sleep(self.WAITING) + def delete_connectivity_service_ODU(self): self.del_serv_input_data["uuid"] = str(tpcetest.uuid_services.odu) response = test_utils.transportpce_api_rpc_request( @@ -666,6 +694,7 @@ def delete_connectivity_service_ODU(self): tpcetest.assertIn(response["status_code"], (requests.codes.ok, requests.codes.no_content)) time.sleep(self.WAITING) + def delete_connectivity_service3_ODU(self): self.del_serv_input_data["uuid"] = str(tpcetest.uuid_services3.odu) response = test_utils.transportpce_api_rpc_request( @@ -673,6 +702,7 @@ def delete_connectivity_service3_ODU(self): tpcetest.assertIn(response["status_code"], (requests.codes.ok, requests.codes.no_content)) time.sleep(self.WAITING) + def delete_connectivity_service_PhotonicMedia(self): self.del_serv_input_data["uuid"] = str(tpcetest.uuid_services.pm) response = test_utils.transportpce_api_rpc_request( @@ -680,6 +710,7 @@ def delete_connectivity_service_PhotonicMedia(self): tpcetest.assertIn(response["status_code"], (requests.codes.ok, requests.codes.no_content)) time.sleep(self.WAITING) + def delete_connectivity_service3_PhotonicMedia(self): self.del_serv_input_data["uuid"] = str(tpcetest.uuid_services3.pm) response = test_utils.transportpce_api_rpc_request( @@ -687,6 +718,7 @@ def delete_connectivity_service3_PhotonicMedia(self): tpcetest.assertIn(response["status_code"], (requests.codes.ok, requests.codes.no_content)) time.sleep(self.WAITING) + def delete_connectivity_service2_PhotonicMedia(self): self.del_serv_input_data["uuid"] = str(tpcetest.uuid_services2.pm) response = test_utils.transportpce_api_rpc_request( @@ -694,6 +726,7 @@ def delete_connectivity_service2_PhotonicMedia(self): tpcetest.assertIn(response["status_code"], (requests.codes.ok, requests.codes.no_content)) time.sleep(self.WAITING) + def get_no_tapi_services(self): response = test_utils.transportpce_api_rpc_request( 'tapi-connectivity', 'get-connectivity-service-list', None) @@ -703,6 +736,7 @@ def get_no_tapi_services(self): "error-message": "No services exist in datastore"}, response['output']['errors']['error']) + def cleanup_and_desinstall_tapi_feature(): disconnect_spdrA_C() disconnect_roadmA_C() @@ -712,6 +746,7 @@ def cleanup_and_desinstall_tapi_feature(): # response = test_utils.get_ordm_serv_list_request() # tpcetest.assertEqual(response['status_code'], requests.codes.conflict) + def disconnect_spdrA_C(): response = test_utils.unmount_device("SPDR-SA1") if response.status_code not in [204]: @@ -720,6 +755,7 @@ def disconnect_spdrA_C(): if response.status_code not in [204]: print("disconnection of SPDR-SC1 failed") + def disconnect_roadmA_C(): response = test_utils.unmount_device("ROADM-A1") if response.status_code not in [204]: @@ -728,6 +764,7 @@ def disconnect_roadmA_C(): if response.status_code not in [204]: print("disconnection of ROADM-C1 failed") + def uninstall_Tapi_Feature(): test_utils.uninstall_karaf_feature("odl-transportpce-tapi") time.sleep(5) @@ -736,16 +773,16 @@ def uninstall_Tapi_Feature(): print("Tapi Feature uninstallation :") # tpcetest.assertEqual(response['status_code'], requests.codes.ok) # print("Tapi Feature uninstalled") - if('node' not in response['network'][0] and 'ietf-network-topology:link' not in response['network'][0]): - print(" otn topology empty") + if ('node' not in response['network'][0] and 'ietf-network-topology:link' not in response['network'][0]): + print(" otn topology empty") # tpcetest.assertNotIn('node', response['network'][0]) # tpcetest.assertNotIn('ietf-network-topology:link', response['network'][0]) response = test_utils.get_ietf_network_request('openroadm-topology', 'config') if ((requests.codes.ok in response['status_code']) and ('node' not in response['network'][0] and 'ietf-network-topology:link' not in response['network'][0])): - # tpcetest.assertEqual(response['status_code'], requests.codes.ok) - # tpcetest.assertNotIn('node', response['network'][0]) - # tpcetest.assertNotIn('ietf-network-topology:link', response['network'][0]) + # tpcetest.assertEqual(response['status_code'], requests.codes.ok) + # tpcetest.assertNotIn('node', response['network'][0]) + # tpcetest.assertNotIn('ietf-network-topology:link', response['network'][0]) print("OpenROADM topology also empty") print("Confirm Tapi Feature correctly uninstalled") diff --git a/tests/transportpce_tests/pce/test05_pce_tapi.py b/tests/transportpce_tests/pce/test05_pce_tapi.py index 7a227491f..fd95587f3 100644 --- a/tests/transportpce_tests/pce/test05_pce_tapi.py +++ b/tests/transportpce_tests/pce/test05_pce_tapi.py @@ -12,6 +12,7 @@ # pylint: disable=no-member # pylint: disable=too-many-public-methods +import test_utils_generate_tapi_topo import os import unittest import time @@ -22,7 +23,6 @@ sys.path.append('transportpce_tests/common/') # pylint: disable=wrong-import-position # pylint: disable=import-error import test_utils # nopep8 -import test_utils_generate_tapi_topo # pylint: disable=too-few-public-methods @@ -33,6 +33,7 @@ class UuidServices: self.odu = None self.dsr = None + class TransportPCEtest(unittest.TestCase): # SIP+SPDR-SA1-XPDR1+PHOTONIC_MEDIA_OTS+XPDR1-NETWORK1 UUID IS sAOTS = "38d81f55-1798-3520-ba16-08efa56630c4" @@ -77,10 +78,10 @@ class TransportPCEtest(unittest.TestCase): "service-rate": "100", "clli": "38c114ae-9c0e-3068-bb27-db2dbd81220b", "service-format": "OTU", - "tx-direction" : { + "tx-direction": { "logical-connection-point": "4e4bf439-b457-3260-865c-db716e2647c2" }, - "rx-direction" : { + "rx-direction": { "logical-connection-point": "4e4bf439-b457-3260-865c-db716e2647c2" }, "node-id": "SPDR-SA1-XPDR2+XPONDER" @@ -89,10 +90,10 @@ class TransportPCEtest(unittest.TestCase): "service-rate": "100", "clli": "d852c340-77db-3f9a-96e8-cb4de8e1004a", "service-format": "OTU", - "tx-direction" : { + "tx-direction": { "logical-connection-point": "ae235ae1-f17f-3619-aea9-e6b2a9e850c3" }, - "rx-direction" : { + "rx-direction": { "logical-connection-point": "ae235ae1-f17f-3619-aea9-e6b2a9e850c3" }, "node-id": "SPDR-SC1-XPDR2+XPONDER" @@ -245,9 +246,8 @@ class TransportPCEtest(unittest.TestCase): # s2ZOTS = "c7549fcc-895f-3fa2-ad81-ea00e25772e9" # # SIP+SPDR-SA1-XPDR3+PHOTONIC_MEDIA_OTS+XPDR3-NETWORK1 UUID IS del_serv_input_data = {"uuid": "TBD"} - - tapi_topo = {"topology-id": "TBD"} + tapi_topo = {"topology-id": "TBD"} @classmethod def setUpClass(cls): @@ -256,22 +256,22 @@ class TransportPCEtest(unittest.TestCase): test_utils_generate_tapi_topo.connect_rdms() test_utils_generate_tapi_topo.interconnect_roadm_and_xpdr() test_utils_generate_tapi_topo.add_omsAttributes_to_ROADM2ROADM_links() - #test_utils_generate_tapi_topo.create_services_on_infra() + # test_utils_generate_tapi_topo.create_services_on_infra() print("Entering PCE Test 05 to test PCE algo for TAPI") @classmethod def tearDownClass(cls): # pylint: disable=not-an-iterable # test_utils_generate_tapi_topo.delete_created_tapi_services(self) - #test_utils_generate_tapi_topo.cleanup_and_desinstall_tapi_feature() + # test_utils_generate_tapi_topo.cleanup_and_desinstall_tapi_feature() print("Exiting PCE Test 05 to test PCE algo for TAPI") - #test_utils_generate_tapi_topo.tearDownClass(cls) + # test_utils_generate_tapi_topo.tearDownClass(cls) def setUp(self): time.sleep(2) - # Test failed path computation + def test_01_fail_path_computation(self): response = test_utils.transportpce_api_rpc_request('transportpce-pce', 'path-computation-request', @@ -281,6 +281,7 @@ class TransportPCEtest(unittest.TestCase): response['output']['configuration-response-common']['response-message']) time.sleep(2) # Path Computation success + def test_02_successfull_path_computation_spdr_x2(self): response = test_utils.transportpce_api_rpc_request('transportpce-pce', 'path-computation-request', @@ -363,7 +364,6 @@ class TransportPCEtest(unittest.TestCase): # response['output']['configuration-response-common']['response-message']) # time.sleep(2) - # # # Test1 success path computation # def test_15_success1_path_computation(self): @@ -375,12 +375,12 @@ class TransportPCEtest(unittest.TestCase): # self.path_computation_input_data["hard-constraints"] = {"customer-code": ["Some customer-code"], # "co-routing": { # "service-identifier-list": [{ - #2 "service-identifier": "Some existing-service"}] + # 2 "service-identifier": "Some existing-service"}] # }} # self.path_computation_input_data["soft-constraints"] = {"customer-code": ["Some customer-code"], # "co-routing": { # "service-identifier-list": [{ - #2 "service-identifier": "Some existing-service"}] + # 2 "service-identifier": "Some existing-service"}] # }} # response = test_utils.transportpce_api_rpc_request('transportpce-pce', # 'path-computation-request', @@ -485,8 +485,8 @@ class TransportPCEtest(unittest.TestCase): # # # def create_connectivity_service2_PhotonicMedia(self): -#2 self.cr_serv_input_data["end-point"][0]["service-interface-point"]["service-interface-point-uuid"] = self.s2AOTS -#2 self.cr_serv_input_data["end-point"][1]["service-interface-point"]["service-interface-point-uuid"] = self.s2ZOTS +# 2 self.cr_serv_input_data["end-point"][0]["service-interface-point"]["service-interface-point-uuid"] = self.s2AOTS +# 2 self.cr_serv_input_data["end-point"][1]["service-interface-point"]["service-interface-point-uuid"] = self.s2ZOTS # self.cr_serv_input_data["end-point"][0]["local-id"] = "SPDR-SA1-XPDR2" # self.cr_serv_input_data["end-point"][1]["local-id"] = "SPDR-SC1-XPDR2" # self.cr_serv_input_data["end-point"][0]["name"][0]["value"] = "SPDR-SA1-XPDR2" @@ -529,6 +529,5 @@ class TransportPCEtest(unittest.TestCase): # time.sleep(1) - if __name__ == "__main__": unittest.main(verbosity=2)